package com.wisdom.client.rpc;

import com.wisdom.client.cache.NacosServiceProvider;
import com.wisdom.client.cache.NacosServiceProviderCache;
import com.wisdom.client.cluster.ClientLoadBalanceStrategy;
import com.wisdom.client.cluster.StrategyProvider;
import com.wisdom.client.config.ClientRpcConfig;
import com.wisdom.client.handler.RpcResponseHandler;
import com.wisdom.core.data.RpcRequest;
import com.wisdom.core.data.RpcResponse;
import com.wisdom.core.exception.RpcException;
import com.wisdom.core.netty.handler.client.RpcRequestEncoder;
import com.wisdom.core.netty.handler.client.RpcResponseDecoder;
import com.wisdom.core.netty.handler.coder.FrameDecoder;
import com.wisdom.core.netty.handler.coder.FrameEncoder;
import com.wisdom.core.netty.request.ChannelMapping;
import com.wisdom.core.netty.request.RequestPromise;
import com.wisdom.core.netty.request.RpcRequestHolder;
import com.wisdom.core.provider.ServiceProvider;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.util.concurrent.DefaultThreadFactory;
import io.netty.util.concurrent.Future;
import io.netty.util.concurrent.GenericFutureListener;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.concurrent.TimeUnit;

/**
 * @description: <Manager all rpc request>
 * @author: Leo
 * @date: 2021-08-23 18:29
 */
@Component
@Slf4j
public class RpcRequestManager {

    @Autowired
    private NacosServiceProviderCache cache;

    @Autowired
    private ClientRpcConfig clientRpcConfig;

    @Autowired
    private StrategyProvider strategyProvider;

    RpcResponseHandler rpcResponseHandler = new RpcResponseHandler();


    public RpcResponse sendRequest(RpcRequest rpcRequest) {
        /*
                  1. 获取需要请求的服务信息集合
                  2. 负载均衡获取单一服务
                  3. Netty调用
         */
        //1. 获取需要请求的服务信息集合
        List<NacosServiceProvider> providers = cache.get(rpcRequest.getClassName());
        if (providers == null || providers.isEmpty()) {
            throw new RpcException("没有发现可以调用的服务");
        }
        //2. 负载均衡获取单一服务
        ClientLoadBalanceStrategy strategy = strategyProvider.getLoadBalanceStrategy();
        ServiceProvider provider = strategy.getProvider(providers);
        //3. Netty调用
        return nettyRequest(provider, rpcRequest);
    }

    private RpcResponse nettyRequest(ServiceProvider provider, RpcRequest rpcRequest) {
        Channel channel;
        try {
            //判定channel是否存在
            if (!RpcRequestHolder.channelExist(provider.getServerIp(), provider.getServerPort())) {
                EventLoopGroup work = new NioEventLoopGroup(0, new DefaultThreadFactory("work-"));
                Bootstrap bootstrap = new Bootstrap();
                bootstrap.group(work)
                        .channel(NioSocketChannel.class)
                        .handler(new ChannelInitializer<SocketChannel>() {
                            @Override
                            protected void initChannel(SocketChannel sc) {
                                ChannelPipeline pipeline = sc.pipeline();
                                //pipeline.addLast(new LoggingHandler(LogLevel.INFO));
                                //发送数据
                                pipeline.addLast("FrameEncoder", new FrameEncoder());
                                pipeline.addLast("RpcRequestEncoder", new RpcRequestEncoder());

                                //接收响应
                                pipeline.addLast("FrameDecoder", new FrameDecoder());
                                pipeline.addLast("RpcResponseDecoder", new RpcResponseDecoder());
                                pipeline.addLast("RpcResponseHandler", rpcResponseHandler);
                            }
                        });
                //创建netty连接
                ChannelFuture future = bootstrap.connect(provider.getServerIp(), provider.getServerPort()).sync();
                if (future.isSuccess()) {
                    channel = future.channel();
                    RpcRequestHolder.addChannelMapping(new ChannelMapping(provider.getServerIp(),
                            provider.getServerPort(), channel));

                } else {
                    log.error("remote rpc request error,msg is {}", future.cause().getMessage());
                    throw new RpcException(future.cause().getMessage(), future.cause());
                }
            } else {
                channel = RpcRequestHolder.getChannel(provider.getServerIp(), provider.getServerPort());
                if (channel == null){
                    throw new RpcException("channelMappingMap get Channel failed");
                }
            }
            RequestPromise promise = new RequestPromise(channel.eventLoop());
            RpcRequestHolder.addRequestPromise(rpcRequest.getRequestId(), promise);
            //连接成功 发送数据
            channel.writeAndFlush(rpcRequest);
            promise.addListener(new GenericFutureListener<Future<? super RpcResponse>>() {
                @Override
                public void operationComplete(Future<? super RpcResponse> future) throws Exception {
                    if (!future.isSuccess()) {
                        RpcRequestHolder.removeRequestPromise(rpcRequest.getRequestId());
                    }
                }
            });
            //等待 RpcResponseHandler 的响应结果
            RpcResponse response = (RpcResponse) promise.get((clientRpcConfig.getConnectTimeout() * 1000L), TimeUnit.SECONDS);
            RpcRequestHolder.removeRequestPromise(response.getRequestId());
            return response;

        } catch (Exception e) {
            log.error("remote rpc request error,msg is {}", e.getMessage());
            throw new RpcException(e.getMessage(), e);
        }
    }


}
